package eventbus

import (
	"context"
	"fmt"
	"sync"

	"gitee.com/linxing_3/sr-core/plugins/log"
	"gitee.com/linxing_3/sr-core/plugins/mq"
	"golang.org/x/sync/errgroup"
)

var (
	DefaultMQBus IEventBusProvider = NewMQBus()
)

func NewMQBus() IEventBusProvider {
	return &mqBus{
		consumer: make(map[string]map[string]SubscribeCallback),
		lock:     sync.Mutex{},
	}
}

type mqBus struct {
	consumer map[string]map[string]SubscribeCallback
	lock     sync.Mutex

	mq mq.IMsgQueue
}

func (e *mqBus) SetMQ(mq mq.IMsgQueue) {
	e.mq = mq
}

// AsyncPublish implements IMQBusProvider.
func (e *mqBus) AsyncPublish(ctx context.Context, msg *Message) error {

	log.WithFields(
		log.NewFiled("name", "eventbus").
			WithField("topic", msg.Topic).
			WithField("wait", msg.Wait).
			WithField("syncToMQIfNoSubs", msg.SyncToMQIfNoSubs),
	).Debug("AsyncPublish")

	return e.asyncConsumeMessage(ctx, msg)
}

// Publish implements IMQBusProvider.
func (e *mqBus) Publish(ctx context.Context, msg *Message) error {
	log.WithFields(
		log.NewFiled("name", "eventbus").
			WithField("topic", msg.Topic).
			WithField("wait", msg.Wait).
			WithField("syncToMQIfNoSubs", msg.SyncToMQIfNoSubs),
	).Debug("AsyncPublish")
	return e.syncConsumeMessage(ctx, msg)
}

// Subscribe implements IMQBusProvider.
func (e *mqBus) Subscribe(topic string, handler SubscribeCallback) (string, error) {
	e.lock.Lock()
	defer e.lock.Unlock()

	key := fmt.Sprintf("%s_%p", topic, handler)
	if container, ok := e.consumer[topic]; ok {
		container[key] = handler
	} else {
		e.consumer[topic] = map[string]SubscribeCallback{
			key: handler,
		}
	}
	return key, nil
}

func (e *mqBus) getSubscribes(topic string) map[string]SubscribeCallback {

	if container, ok := e.consumer[topic]; ok {
		return container
	}
	return nil
}

// 同步执行订阅消息
func (e *mqBus) syncConsumeMessage(ctx context.Context, msg *Message) error {

	container := e.getSubscribes(msg.Topic)
	if container == nil {
		return nil
	}

	for _, handler := range container {
		if err := handler(ctx, msg); err != nil {
			return err
		}
	}

	if msg.SyncToMQIfNoSubs && (e.mq != nil || msg.MQ != nil) {
		mqPlugin := e.mq
		if msg.MQ != nil {
			mqPlugin = msg.MQ
		}

		msg, err := mq.NewMQMsgNowJson(msg.Topic, msg.Body)
		if err != nil {
			return err
		}
		return mqPlugin.Publish(ctx, msg)
	}

	return nil
}

// 异步执行订阅消息
func (e *mqBus) asyncConsumeMessage(ctx context.Context, msg *Message) error {

	container := e.getSubscribes(msg.Topic)
	if container == nil {
		return nil
	}
	if msg.Wait {
		wg, ctx := errgroup.WithContext(ctx)
		for _, handler := range container {
			e.wgConsume(wg, handler, ctx, msg)
		}
		return wg.Wait()
	} else {

		for _, handler := range container {
			go handler(ctx, msg)
		}
	}
	return nil
}

func (e *mqBus) wgConsume(wg *errgroup.Group, handler SubscribeCallback, ctx context.Context, msg *Message) {
	wg.Go(func() error {
		return handler(ctx, msg)
	})
}
